package product.utils;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.text.DecimalFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

import cn.binarywang.tools.generator.ChineseMobileNumberGenerator;
import cn.binarywang.tools.generator.ChineseNameGenerator;
import joinery.DataFrame;

/**
 * 数据生产工具
 */
public class DataUtils {

	/**
	 * 随机生成姓名、电话号
	 */
	public static void geneNamNum() {
		// 准备生成器和df对象
		ChineseNameGenerator cng = ChineseNameGenerator.getInstance();
		ChineseMobileNumberGenerator cmng = ChineseMobileNumberGenerator.getInstance();
		DataFrame<Object> df = new DataFrame<>("name", "telephone");
		for (int i = 0; i < 10; i++) {
			//生成中文姓名并放入df
			String generatedName = cng.generate();
			String generatedMobileNum = cmng.generate();
			System.out.println(generatedName + "," + generatedMobileNum);
			df.append(Arrays.asList(generatedName, generatedMobileNum));
		}
		try {
			// 生成csv文件到运行环境的当前目录下
			df.writeCsv("./nam_num.txt");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 时间维度表
	 */
	public static void genDateDimention() {
		DataFrame<Object> df = new DataFrame<>("year", "month", "day");
		Calendar calendar = Calendar.getInstance();
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
		try {
			//设置时间段的起止时间
			Date start = sdf.parse("2017-01-01");
			Date end = sdf.parse("2019-12-31");
			calendar.setTime(start);
			// 先添加年月日
			while (end.compareTo(calendar.getTime()) > 0) {
				System.out.println(calendar.get(Calendar.YEAR) + "," + (calendar.get(Calendar.MONTH) + 1) + ","
						+ calendar.get(Calendar.DATE));
				// 注意月份从0开始，需要+1
				df.append(Arrays.asList(calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1,
						calendar.get(Calendar.DATE)));
				// 日期+1
				calendar.add(Calendar.DATE, 1);
			}
			// 再添加年维度和月维度的，缺失的维度设为-1
			for (int year = 2017; year < 2020; year++) {
				df.append(Arrays.asList(year, -1, -1));
				for (int month = 1; month <= 12; month++) {
					df.append(Arrays.asList(year, month, -1));
				}
			}
			// 生成csv文件
			df.writeCsv("./dateDimentions.txt");
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 生成一定范围内的长整数
	 */
	public static Long nextLong(Random rng, Long n) {
		long bits, val;
		do {
			bits = (rng.nextLong() << 1) >>> 1;
			val = bits % n;
		} while (bits - val + (n - 1) < 0L);
		return val;
	}

	/**
	 * 生成一段时间内的时间戳
	 */
	public static Long getTimeStamp(SimpleDateFormat formatter, String startStr, String endStr) {
		Random random = new Random();
		Long startTime = null, endTime = null, offset = null;
		try {
			startTime = formatter.parse(startStr).getTime();
			endTime = formatter.parse(endStr).getTime();
			//这里就用到了之前的nextLong，偏移的范围在起止时间相差的时间段内
			offset = nextLong(random, endTime - startTime);
		} catch (ParseException e) {
			e.printStackTrace();
		}
		return startTime + offset;
	}

	/**
	 * 读取之前的nam_num.txt文件并放入Map，用于随机抽取通话人
	 */
	public static Map<String, String> getNamNumMap(String pathname) {
		Map<String, String> map = new HashMap<String, String>();
		//Java7的try-with-resources可以优雅关闭文件，异常时自动关闭文件
		try (FileReader reader = new FileReader(pathname); BufferedReader br = new BufferedReader(reader) // 建立一个对象，它把文件内容转成计算机能读懂的语言
		) {
			String line = br.readLine();
			String[] strs;
			while ((line = br.readLine()) != null) {
				// 一次读入一行数据
				strs = line.split(",");
				map.put(strs[0], strs[1]);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return map;
	}

	/**
	 * 写txt文件
	 */
	public static void writeRecord(String outputFilePath, String line) {
		// 构造好记录后就往文件中写，输出的文件通过主函数的参数传递进来
		try (FileWriter writer = new FileWriter(outputFilePath, true); BufferedWriter br = new BufferedWriter(writer)) {
			br.write(line + "\n");
			br.flush();
			//设置休眠时间来调整，数据生产的速度
			Thread.sleep(10 * 1000);
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 构造rowKey
	 * <p>
	 * 这是技巧性非常强的地方，讲究整体分散，局部聚合。
	 * <p>
	 * 首先整体分散，需要制定合理的分区规则进行分区，避免热点问题，我这里用的规则是
	 * <p>
	 * int pid = str.hashCode()&Integer.MAX_VALUE % 3;这样得到的结果应该能较为均匀的分在HBase的三个分区当中
	 * <p>
	 * 其次局部聚合，最好能通过指定起止位置来获得所有想要的数据，避免全表扫描。
	 * <p>
	 * 这就决定着上面的str如何选取了，我这里选的是主叫人拼接上日期字符串，例如18481678295_2019-07-11,这样如果我想知道某人一段时间内的通话信息，就可以用其手机号拼接上起止时间字符串作为起止位置来得到数据。
	 */
	public static String genRowKey(String[] values) {
		String dateStr = values[4];
		String str = values[0] + "_" + dateStr.substring(0, 7);
		int pid = str.hashCode() & Integer.MAX_VALUE % 3;
		DecimalFormat df = new DecimalFormat("00");
		String startZeroStr = df.format(pid);
		return startZeroStr + "_" + values[0] + "_" + dateStr.substring(0, 10) + "_" + values[2];
	}

	public static void data() {
		String pathname = "./nam_num.txt";
		String outputFilePath = "./data.txt";

		// 获得基础的姓名手机号映射
		Map<String, String> map = getNamNumMap(pathname);
		String[] keys = new String[map.size()];
		// 获得Key的数组用于随机取出通话人
		map.keySet().toArray(keys);

		SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Random random = new Random();
		String key1, key2, value1, value2;
		while (true) {
			//获得主叫人被叫人姓名
			key1 = keys[random.nextInt(keys.length)];
			key2 = keys[random.nextInt(keys.length)];
			//自己不能够自己打电话
			if (key1 == key2)
				continue;
			//获得主叫人被叫人手机号
			value1 = map.get(key1);
			value2 = map.get(key2);
			//获得通话时间
			Long timeStamp = getTimeStamp(formatter, "2017-01-01 00:00:00", "2019-07-11 23:59:59");
			//获得通话时间
			int callTime = random.nextInt(9999);
			String line = key1 + "," + value1 + "," + key2 + "," + value2 + "," + String.valueOf(timeStamp) + ","
					+ String.valueOf(callTime);
			System.out.println(line);
			writeRecord(outputFilePath, line);
		}
	}

	/**
	 * 构造记录了，注意主函数需要两个参数，一个是nam_num.txt文件路径另一个是生产记录的txt文件
	 */
	public static void main(String[] args) {
		data();
	}

}
